package com.hzy.es.service;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.hzy.es.entity.EsSearchEntity;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.ElasticsearchException;
import org.elasticsearch.action.DocWriteRequest;
import org.elasticsearch.action.DocWriteResponse;
import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.get.*;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.support.master.AcknowledgedResponse;
import org.elasticsearch.action.support.replication.ReplicationResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.client.indices.CreateIndexResponse;
import org.elasticsearch.client.indices.GetIndexRequest;
import org.elasticsearch.client.indices.PutMappingRequest;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.reindex.BulkByScrollResponse;
import org.elasticsearch.index.reindex.DeleteByQueryRequest;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.fetch.subphase.FetchSourceContext;
import org.springframework.beans.factory.annotation.Autowired;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

@Slf4j
public class EsSearchServiceImp {
    @Autowired
    private RestHighLevelClient restHighLevelClient;

    public List<Map<String, Object>> searchResult(SearchRequest searchRequest) {
        try {
            SearchResponse searchResponse = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
            RestStatus status = searchResponse.status();
            log.info("状态：" + status);
            TimeValue took = searchResponse.getTook();
            log.info("执行时间：" + took);
            Boolean terminatedEarly = searchResponse.isTerminatedEarly();
            log.info("请求终止：" + terminatedEarly);
            boolean timedOut = searchResponse.isTimedOut();
            log.info("超时时间：" + timedOut);
            //获得响应的文档
            SearchHits hits = searchResponse.getHits();
            //迭代取出数据
            for (SearchHit hit : hits) {
                Map<String, Object> sourceAsMap = hit.getSourceAsMap();
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }

    /**
     * 存在
     *
     * @throws Exception
     */
    public boolean exists(String indexName, String id) throws Exception {
        GetRequest getRequest = new GetRequest(indexName, id);
        getRequest.fetchSourceContext(new FetchSourceContext(false));
        getRequest.storedFields("_none_");

        //Synchronous Execution
        boolean exists = restHighLevelClient.exists(getRequest, RequestOptions.DEFAULT);
        return exists;
    }

    /**
     * 根据 id 获取数据
     *
     * @throws Exception
     */
    public void get(String indexName, String id) throws Exception {
        GetRequest request = new GetRequest(indexName, id);

        //可以自定义要查询的具体key
        /*request.fetchSourceContext(FetchSourceContext.DO_NOT_FETCH_SOURCE);
        String[] includes = new String[]{"name", "price"};
        String[] excludes = Strings.EMPTY_ARRAY;
        FetchSourceContext fetchSourceContext = new FetchSourceContext(true, includes, excludes);
        request.fetchSourceContext(fetchSourceContext);*/

        //Synchronous Execution
        GetResponse getResponse = restHighLevelClient.get(request, RequestOptions.DEFAULT);

        //Get Response
        String index = getResponse.getIndex();
        String type = getResponse.getType();
        String ID = getResponse.getId();
        if (getResponse.isExists()) {
            long version = getResponse.getVersion();
            String sourceAsString = getResponse.getSourceAsString();
            System.out.println("index: " + index);
            System.out.println("type: " + type);
            System.out.println("id: " + ID);
            System.out.println("version: " + version);
            System.out.println(sourceAsString);
        } else {
            System.out.println("没有查询到结果");
        }
    }

    /**
     * 删除数据
     *
     * @param indexName
     * @param id
     * @throws IOException
     */
    public void delete(String indexName, String id) throws IOException {
        DeleteRequest request = new DeleteRequest(indexName, id);
        //Synchronous Execution
        DeleteResponse deleteResponse = restHighLevelClient.delete(request, RequestOptions.DEFAULT);

        // document was not found
        if (deleteResponse.getResult() == DocWriteResponse.Result.NOT_FOUND) {
            System.out.println("要删除的数据不存在");
        } else {
            //Delete Response
            String index = deleteResponse.getIndex();
            String type = deleteResponse.getType();
            String ID = deleteResponse.getId();
            long version = deleteResponse.getVersion();
            System.out.println("index: " + index);
            System.out.println("type: " + type);
            System.out.println("id: " + ID);
            System.out.println("version: " + version);
            System.out.println("status: " + "DELETE");
            ReplicationResponse.ShardInfo shardInfo = deleteResponse.getShardInfo();
            if (shardInfo.getFailed() > 0) {
                for (ReplicationResponse.ShardInfo.Failure failure : shardInfo.getFailures()) {
                    String reason = failure.reason();
                }
            }
        }
    }

    /**
     * 删除数据 -- 批量删除
     *
     * @param queryBuilder
     * @throws IOException
     */
    public void delete(QueryBuilder queryBuilder, String... index) throws IOException {
        DeleteByQueryRequest deleteByQueryRequest = new DeleteByQueryRequest(index);
        deleteByQueryRequest.setQuery(queryBuilder);
        BulkByScrollResponse bulkByScrollResponse = restHighLevelClient.deleteByQuery(deleteByQueryRequest, RequestOptions.DEFAULT);
        bulkByScrollResponse.getDeleted();
    }

    /**
     * 判断index是否存在
     *
     * @param index
     * @return
     */
    public boolean checkIndex(String index) {
        try {
            GetIndexRequest request = new GetIndexRequest(index);
            boolean result = restHighLevelClient.indices().exists(request, RequestOptions.DEFAULT);
            return result;
        } catch (IOException e) {
            e.printStackTrace();
            return false;
        }
    }

    /**
     * 创建index
     *
     * @param index
     * @return
     */
    public Boolean createIndex(String index) throws IOException {
        CreateIndexRequest createIndexRequest = new CreateIndexRequest(index);
        createIndexRequest.source("", XContentType.CBOR);
        CreateIndexResponse createIndexResponse = restHighLevelClient.indices().create(createIndexRequest, RequestOptions.DEFAULT);
        if (createIndexResponse.isAcknowledged()) {
            return true;
        }
        return false;
    }

    /**
     * 创建mapping
     * 注意数据格式，此版本已经取去除String格式，改为text和keyword格式，其中text格式支持分词和建立索引，
     * 支持模糊查询和精确查询，不支持聚合，keyword不支持分词，支持模糊查询和精确查询，支持聚合查询，排序
     *
     * @param index
     * @param builder
     * @return
     * @throws IOException
     */
    public Boolean createMapping(String index, XContentBuilder builder) throws IOException {
        PutMappingRequest putMappingRequest = new PutMappingRequest(index);
        putMappingRequest.source(builder);
        AcknowledgedResponse putMappingResponse = restHighLevelClient.indices().putMapping(putMappingRequest, RequestOptions.DEFAULT);
        if (putMappingResponse.isAcknowledged()) {
            return true;
        }
        return false;
    }

    /**
     * 根据id批量获取数据
     *
     * @throws Exception
     */
    public void multiGet(String indexName, String... ids) throws Exception {
        MultiGetRequest request = new MultiGetRequest();
        for (String str : ids) {
            request.add(new MultiGetRequest.Item(indexName, str));
        }
        MultiGetResponse responses = restHighLevelClient.mget(request, RequestOptions.DEFAULT);

        //Multi Get Response
        MultiGetItemResponse[] Items = responses.getResponses();
        for (MultiGetItemResponse Item : Items) {
            GetResponse response = Item.getResponse();
            String index = response.getIndex();
            String type = response.getType();
            String id = response.getId();
            if (response.isExists()) {
                long version = response.getVersion();
                String sourceAsString = response.getSourceAsString();
                System.out.println("index: " + index);
                System.out.println("type: " + type);
                System.out.println("id: " + id);
                System.out.println("version: " + version);
                System.out.println(sourceAsString + "\n");
            } else {
                System.out.println("不存在");
            }
        }
    }

    public void update(String indexName, String id, String jsonStr) throws Exception {
        try {
            UpdateRequest request = new UpdateRequest(indexName, id);
            request.doc(jsonStr, XContentType.JSON);
            //Synchronous Execution
            UpdateResponse updateResponse = restHighLevelClient.update(request, RequestOptions.DEFAULT);

            //update Response
            String index = updateResponse.getIndex();
            String ID = updateResponse.getId();
            long version = updateResponse.getVersion();
            if (updateResponse.getResult() == DocWriteResponse.Result.CREATED) {
                System.out.println("index: " + index);
                System.out.println("id: " + ID);
                System.out.println("version: " + version);
                System.out.println("status: " + "CREATED");
            } else if (updateResponse.getResult() == DocWriteResponse.Result.UPDATED) {
                System.out.println("index: " + index);
                System.out.println("id: " + ID);
                System.out.println("version: " + version);
                System.out.println("status: " + "UPDATED");
            } else if (updateResponse.getResult() == DocWriteResponse.Result.DELETED) {
                System.out.println("index: " + index);
                System.out.println("id: " + ID);
                System.out.println("version: " + version);
                System.out.println("status: " + "DELETED");
            } else if (updateResponse.getResult() == DocWriteResponse.Result.NOOP) {
                System.out.println("index: " + index);
                System.out.println("id: " + ID);
                System.out.println("version: " + version);
                System.out.println("status: " + "NOOP");
            }
        } catch (ElasticsearchException e) {
            if (e.status() == RestStatus.NOT_FOUND) {
                System.out.println("要修改的内容不存在");
            }
        }
    }

    /**
     * 增，改数据
     *
     * @param indexName index名字
     * @param jsonStr   增加或修改的数据json字符串格式
     * @throws Exception
     */
    public void index(String indexName, String jsonStr) throws Exception {
        IndexRequest request = new IndexRequest(indexName);

        request.source(jsonStr, XContentType.JSON);
        IndexResponse indexResponse = restHighLevelClient.index(request, RequestOptions.DEFAULT);//同步

        String index = indexResponse.getIndex();
        String type = indexResponse.getType();
        String ID = indexResponse.getId();
        long version = indexResponse.getVersion();
        if (indexResponse.getResult() == DocWriteResponse.Result.CREATED) {
            System.out.println("index: " + index);
            System.out.println("type: " + type);
            System.out.println("id: " + ID);
            System.out.println("version: " + version);
            System.out.println("status: " + "CREATED");
        } else if (indexResponse.getResult() == DocWriteResponse.Result.UPDATED) {
            System.out.println("index: " + index);
            System.out.println("type: " + type);
            System.out.println("id: " + ID);
            System.out.println("version: " + version);
            System.out.println("status: " + "UPDATED");
        }
        ReplicationResponse.ShardInfo shardInfo = indexResponse.getShardInfo();
        if (shardInfo.getFailed() > 0) {
            for (ReplicationResponse.ShardInfo.Failure failure : shardInfo.getFailures()) {
                String reason = failure.reason();
                System.out.println(reason);
            }
        }
    }

    /**
     * 将查询出的数据通过实体类映射到es中
     * 注意：字段映射必须和es中的一致，可以直接映射，也可以转换为json映射
     *
     * @param indexName
     * @param esSearchList
     * @throws IOException
     */
    public void bulk(String indexName, List<EsSearchEntity> esSearchList) throws IOException {
        BulkRequest bulkRequest = new BulkRequest();
        for (EsSearchEntity esSearch : esSearchList) {
            bulkRequest.add(new IndexRequest(indexName).source(esSearch));
        }
        bulkRequest.timeout("10m");
        BulkResponse bulkResponse = restHighLevelClient.bulk(bulkRequest, RequestOptions.DEFAULT);
        System.out.println(bulkResponse);
        for (BulkItemResponse bulkItemResponse : bulkResponse) {
            DocWriteResponse itemResponse = bulkItemResponse.getResponse();
            if (bulkItemResponse.getOpType() == DocWriteRequest.OpType.INDEX
                    || bulkItemResponse.getOpType() == DocWriteRequest.OpType.CREATE) {
                // 新增
                IndexResponse indexResponse = (IndexResponse) itemResponse;
                String index = indexResponse.getIndex();
                String type = indexResponse.getType();
                String id = indexResponse.getId();
                long version = indexResponse.getVersion();
                System.out.println("index: " + index);
                System.out.println("type: " + type);
                System.out.println("id: " + id);
                System.out.println("version: " + version);
                System.out.println("status: " + "CREATED" + "\n");
            } else if (bulkItemResponse.getOpType() == DocWriteRequest.OpType.UPDATE) {
                // 修改
                UpdateResponse updateResponse = (UpdateResponse) itemResponse;
                String index = updateResponse.getIndex();
                String type = updateResponse.getType();
                String id = updateResponse.getId();
                long version = updateResponse.getVersion();
                System.out.println("index: " + index);
                System.out.println("type: " + type);
                System.out.println("id: " + id);
                System.out.println("version: " + version);
                System.out.println("status: " + "UPDATE" + "\n");
            } else if (bulkItemResponse.getOpType() == DocWriteRequest.OpType.DELETE) {
                // 删除
                DeleteResponse deleteResponse = (DeleteResponse) itemResponse;
                String index = deleteResponse.getIndex();
                String type = deleteResponse.getType();
                String id = deleteResponse.getId();
                long version = deleteResponse.getVersion();
                System.out.println("index: " + index);
                System.out.println("type: " + type);
                System.out.println("id: " + id);
                System.out.println("version: " + version);
                System.out.println("status: " + "DELETE" + "\n");
            }
        }
    }

    /**
     * 目前es的sql语句还不支持分页查询，如果需要分页查询，请用此方法
     * @param queryBuilder 查询条件builder需要构造好
     * @param searchRequest 查询索引需要构造好
     * @param clazz
     * @param <T>
     * @return
     * @throws IOException
     */
    public <T> List<T> pageQuery(int currentPage, int pageSize, QueryBuilder queryBuilder, SearchRequest searchRequest, Class<T> clazz) throws IOException {

        JSONObject resJSON = new JSONObject();
        JSONArray jsonArr = new JSONArray();
        //最大不能超过 10000条，可以通过设置索引的"max_result_window": "10000000"来解决，不过效率不知如何
        int fromSize = currentPage*pageSize;
        // 创建查询函数构造对象
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();

        sourceBuilder.query(queryBuilder); // 把父查询对象放入函数构造对象中
        sourceBuilder.from(fromSize); // 参数范围起
        sourceBuilder.size(pageSize); // 参数范围始
        sourceBuilder.timeout(new TimeValue(60, TimeUnit.SECONDS));// 设置超时时间
        sourceBuilder.trackTotalHits(true); // 取消默认最大查询数量上限（默认10000）

        searchRequest.source(sourceBuilder);// 把查询函数构造对象注入查询请求中
        // 创建响应对象
        SearchResponse searchResponse = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
        SearchHits searchHits =  searchResponse.getHits();//获取响应中的列表数据
        long total = searchHits.getTotalHits().value;//获取响应中的列表数据总数
        List<T> contents = new ArrayList<>();

        for(SearchHit hit:searchHits.getHits()){// 遍历构造返回JSON，以下不再多说
            JSONObject dataJSON = new JSONObject();
            String tempRes = hit.getSourceAsString();
            dataJSON = JSONObject.parseObject(tempRes);
            jsonArr.add(dataJSON);
            contents.add(JSON.parseObject(tempRes,clazz));
        }

        return contents;

    }

    /**
     * 目前es的sql语句还不支持分页查询，如果需要分页查询，请用此方法
     * @param queryBuilder 查询条件builder需要构造好
     * @param clazz
     * @param indices
     * @param <T>
     * @return
     * @throws IOException
     */
    public <T> List<T> pageQuery(QueryBuilder queryBuilder, Class<T> clazz, String... indices) throws IOException{
        SearchRequest searchRequest = new SearchRequest(indices);
        return this.pageQuery(1, 100,queryBuilder,searchRequest,clazz);
    }
}
